
//// Written by David Goffredo
//// 	June 9, 2009

#ifndef SYNC_FUNC_HPP
#define SYNC_FUNC_HPP

//class sync_func
///////////////////////
//// Generalizes some algorithms used
//// among synchronized threads.
////
//// Common applications would be to have each thread
////	get the min, max, or average of values among
////	the other threads. Any algorithm may be employed.
////
////
//// In addition to each thread performing a binary function,
////   the user may specify a unary function to have the final
////   thread apply to the result before returning the result.
//// The default unary function is 'do_nothing', but one is
////   needed, for example, to compute an average without using
////   a running average, or to compute a median or mode.
//// See end of file for example uses.

#include <boost/thread/mutex.hpp>
#include <boost/thread/barrier.hpp>

////////// some Binary functors  ////////////////////

template<class T>           //static variable, thread-specific variable
struct thread_max {  void operator()(T & stat,const T & thrd) { if(thrd > stat) stat = thrd; }  };

template<class T>
struct thread_min {  void operator()(T & stat,const T & thrd) { if(thrd < stat) stat = thrd; }  };

template<class T>
struct thread_sum {  void operator()(T & stat,const T & thrd) { stat += thrd; }  };

////////// some Unary functors  /////////////////////

template<class T>
struct div_by_n
{
	div_by_n(const T & _n) : n(_n) {}
	void operator()(T & t) { t /= n; }
	private: T n;
};

template<class T>
struct do_nothing { void operator()(T & t) {} };

////////// Class sync_func  ////////////////////

template<class T, class T_binary_ref_func, class T_unary_once_func = do_nothing<T> >
class sync_func
{
		T working;
		const T starting_value;
		T returnable;
		T_binary_ref_func func;  //func( working, thread_piece );
		T_unary_once_func at_end;  //at_end( working );
		boost::mutex moo;
		boost::barrier bar;
		unsigned int n_threads;
		unsigned int counter;
	public:
		sync_func(unsigned int _n_threads, const T & _starting_value,
				  const T_unary_once_func & _at_end = T_unary_once_func(), const T_binary_ref_func & _func = T_binary_ref_func())
			: working(_starting_value), starting_value(_starting_value), func(_func),
			   at_end(_at_end), bar(_n_threads), n_threads(_n_threads), counter(0) {}
		T operator()(const T & thread_piece)
		{
			{
				boost::mutex::scoped_lock lock(moo);
				func( working, thread_piece );
				if(++counter == n_threads)
				{
					at_end( working );
					returnable = working;
					working = starting_value;
					counter = 0;
				}
			}
			bar.wait();
			return returnable;
    	}
};

///////////// Example uses /////////////////////////
//
// namespace Main
// {
//    sync_func< int, thread_max<int> > sync_max(
//                                 nthreads, std::numeric_limits<int>::min() );
//    sync_func< double, thread_sum<double>, div_by_n<double> > sync_avg(
//                                 nthreads, 0, div_by_n(nthreads) );
// }
//
// //... later, in some multi-threaded function ...//
//
// int biggest_in_this_thread = std::max_element(my_portion.beg(), my_portion.end());
// int biggest_in_all_threads = Main.sync_max(biggest_in_this_thread);
// int sum_in_this_thread = std::accumulate(my_portion.beg(), my_portion.end());
// double avg_in_this_thread = sum_in_this_thread / static_cast<double>(my_portion.size());
// double avg_in_all_threads = Main.sync_avg(avg_in_this_thread);
//
/////////////////////////////////////////////////////

#endif //SYNC_FUNC_HPP

